package com.huawei.dli.flink.sye;

import com.huawei.dli.flink.sye.config.FlinkOggMetaConfig;
import com.huawei.dli.flink.sye.sink.HBaseSinkString;
import com.huawei.dli.flink.sye.source.KafkaKerbosSource;
import com.huawei.dli.flink.sye.utils.KafkaUtil;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class App2HBaseString {
    public static void main(String[] args) throws Exception {
        // 获取参数
        ParameterTool params = ParameterTool.fromArgs(args);

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(5000);
        env.setParallelism(5);

        String userPrincipal = params.getRequired("principal");
        KafkaKerbosSource<String> kafkaStringSource = KafkaUtil.getOggKafkaStringSource(params,userPrincipal);
        // 根据topic解析配置
        String topic = params.getRequired("topic");
        FlinkOggMetaConfig flinkOggMetaConfig=new FlinkOggMetaConfig();
        flinkOggMetaConfig.analyzeConfig(topic);

        env.addSource(kafkaStringSource,"kafka_source").disableChaining()
                .addSink(new HBaseSinkString(userPrincipal,flinkOggMetaConfig)).disableChaining();

        env.execute("bigdatadws");
    }
}
